package com.wanlian.socketserver.Device.worker;

import com.wanlian.common.KeyPara;
import com.wanlian.json.JsonStringUtil;
import com.wanlian.netty.model.*;
import com.wanlian.netty.msg.*;
import com.wanlian.socketserver.Device.DeviceOnlineChecker;
import com.wanlian.socketserver.Device.DeviceServer;
import com.wanlian.util.StringUtil;
import io.netty.channel.ChannelId;

import java.util.List;
import java.util.concurrent.*;

public class DeviceServerWorker {
    public static SessionPool sessionPool = new SessionPool();
    private static DeviceServer server;


    /**
     * 工作线程
     */
    public static void work() {
        calcPeerMsg();
        onLineWork();
        DeviceShareSizeWorker.work();
    }


    public static void onLineWork() {
        BlockingQueue<KeyPara> queue = new LinkedBlockingQueue<KeyPara>(10);
        OnlineWorker onlineWorker1 = new OnlineWorker(queue);
        PostWorker postWorker1 = new PostWorker(queue);
        Thread t1 = new Thread(onlineWorker1);
        Thread c1 = new Thread(postWorker1);
        t1.start();
        c1.start();
    }


    public static void replyRegister(String key) {
        RegisterRspMsg registerRspMsg = new RegisterRspMsg();
        registerRspMsg.setClientId(server.getPublicKey());
        registerRspMsg.setTagClientId(key);
        sendDataByKey(key, registerRspMsg);
    }


    /**
     * 发送数据给某个节点
     *
     * @param key
     * @param baseMsg
     */
    public static ResultInfo sendDataByKey(String key, BaseMsg baseMsg) {
        if (sessionPool.isExist(key)) {
            Session session = sessionPool.getSessionByPublicKey(key);
            PeerMsg peerMsg = new PeerMsg(baseMsg);
            session.getChannel().writeAndFlush(peerMsg);
            return new ResultInfo(true);
        } else {
            //  System.out.println("device key:" + key + "终端不存在或已经掉线");
            DeviceOnlineChecker.setOffline(key);
            return new ResultInfo(false, "终端不存在或已经掉线");
        }
    }


    /**
     * 移除
     *
     * @param channelId
     */
    public static void removeByChannelId(ChannelId channelId) throws Exception {
        String key = sessionPool.removeSessionByChannelId(channelId);
        DeviceOnlineChecker.setOffline(key);
    }

    /**
     * 处理网络消息
     */
    private static void calcPeerMsg() {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            fixedThreadPool.execute(new Runnable() {
                public void run() {
                    while (true) {
                        try {
                            TimeUnit.MILLISECONDS.sleep(10);
                            processMsg();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
    }

    /***
     * 处理消息
     */
    private static synchronized void processMsg() {
        try {
            List<PeerMsg> peerMsgs = PeerMsgQueue.get(1000);
            for (PeerMsg peerMsg : peerMsgs) {
                if (peerMsg != null) {
                    String clientId = peerMsg.getBaseMsg().getClientId();
                    String tagClientId = peerMsg.getBaseMsg().getTagClientId();
                    if (StringUtil.hasValue(clientId)) {
                        //当刷新它的时间时，需要修改在线时长
                        if (sessionPool.publicKeyChannelMap.containsKey(clientId)) {
                            sessionPool.publicKeyChannelMap.get(clientId).refrushTime();
                        }
                    }
                    //不填写服务器id
                    if (StringUtil.hasValue(tagClientId) == false) {
                        refrushPing(peerMsg);
                        return;
                    }
                    //填写服务器id
                    if (tagClientId.equals(server.getPublicKey())) {
                        refrushPing(peerMsg);
                        DeviceShareSizeWorker.tidyDeviceShareSize(peerMsg);
                        return;
                    } else {
                        //app->路由器 或者 路由器-->app
                        ResultInfo info = sendDataByKey(peerMsg.getBaseMsg().getTagClientId(), peerMsg.getBaseMsg());
                        if (!info.isBresult()) {
                            DeviceOnlineChecker.setOffline(peerMsg.getBaseMsg().getTagClientId());
                            ResultMsg resultMsg = new ResultMsg();
                            resultMsg.setReplyType(peerMsg.getBaseMsg().getType());
                            resultMsg.setClientId(server.getPublicKey());
                            resultMsg.setTagClientId(peerMsg.getBaseMsg().getClientId());
                            resultMsg.setResultInfo(info);
                            sendDataByKey(resultMsg.getTagClientId(), resultMsg);
                        } else {
                            System.out.println("send success to key:" + peerMsg.getBaseMsg().getTagClientId());
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private static void refrushPing(PeerMsg peerMsg) {
        String msgType = peerMsg.getBaseMsg().getType();
        if (msgType.equals(MsgType.PING)) {
            Address address = JsonStringUtil.readValue(peerMsg.getBaseMsg().getContent(), Address.class);
            if (address != null) {
                sessionPool.changeAddress(address);
            }
            PingReply pingReply = new PingReply();
            pingReply.setClientId(server.getPublicKey());
            pingReply.setTagClientId(peerMsg.getBaseMsg().getClientId());
            sendDataByKey(pingReply.getTagClientId(), pingReply);
        }
    }

    public static DeviceServer getServer() {
        return server;
    }

    public static void setServer(DeviceServer server) {
        DeviceServerWorker.server = server;
    }
}

